package com.netty.handler;

import java.net.SocketAddress;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.netty.common.protocol.RpcRequest;
import com.netty.common.protocol.RpcResponse;
import com.netty.future.RPCFuture;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse>{

	private static final Logger logger = LoggerFactory.getLogger(RpcClientHandler.class);
	
	private ConcurrentHashMap<String, RPCFuture> pendingRPC = new ConcurrentHashMap<String, RPCFuture>();
	
	private volatile Channel channel;
	private SocketAddress remotePeer;
	
	public Channel getChannel() {
		return channel;
	}
	
	public SocketAddress getRemotePeer() {
		return remotePeer;
	}
	
	/**
	 * 活跃的channel
	 */
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		super.channelActive(ctx);
		this.remotePeer = this.channel.remoteAddress();
	}
	/**
	 * 注册过的channel
	 */
	@Override
	public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
		super.channelRegistered(ctx);
		this.channel = ctx.channel();
	}
	/**
	 * 读取数据
	 */
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, RpcResponse response)
			throws Exception {
		String requestId = response.getRequestId();
		RPCFuture rpcFuture = pendingRPC.get(requestId);
		if (rpcFuture != null) {
			pendingRPC.remove(requestId);
			rpcFuture.done(response);
		}
		
	}
	/**
	 * 捕捉异常
	 */
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
			throws Exception {
		logger.error("client caught exception",cause);
		ctx.close();
	}

	public void close(){
		channel.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
	}
	
	public RPCFuture sendRequest(RpcRequest request){
		final CountDownLatch latch = new CountDownLatch(1);
		RPCFuture rpcFuture = new RPCFuture(request);
		pendingRPC.put(request.getRequestId(), rpcFuture);
		channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
			
			@Override
			public void operationComplete(ChannelFuture future) throws Exception {
				latch.countDown();
			}
		});
		try {
			latch.await();
		} catch (Exception e) {
			logger.error("",e);
		}
		return rpcFuture;
	}
	
	
	
	
}
